package com.bingxu.flink.chapter02.sink;

import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;

/**
 * @author :BingXu
 * @description :TODO
 * @date :2021/8/10 11:48
 * @modifier :
 */

public class Trans_KafkaSink {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> streamDS = env.socketTextStream("localhost", 9999);

        // sink funcation
        FlinkKafkaProducer<String> kafkaProducer = new FlinkKafkaProducer<>(
                "hadoop102:9092,hadoop103:9092,hadoop104:9092",
                "flink210323",
                new SimpleStringSchema());
        streamDS.addSink(kafkaProducer);

        env.execute();
    }
}
